package com.apps.sdses.flink141.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description 模拟水位数据写入Kafka
 * @date 2023/6/12 9:02
 */
public class Datagen2Kafka_WaterSensor {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id       BIGINT,\n" +     //采集信息id
                "    sid      STRING,\n" +    //水位传感器id
                "    ts   TIMESTAMP(3),\n" +    //数据采集时间
                "    flag   BOOLEAN,\n" +       //水位报警开关
                "    vc INT\n" +                //采集到的水位高度
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
                "  'number-of-rows' = '5', \n" +   //一次生成100万条数据
//                "  'rows-per-second' = '1', \n" +   //每秒生成10000条数据
                "  'fields.sid.length' = '5', \n" +   //传感器id长度
                "  'fields.id.min' = '100', \n" +
                "  'fields.id.max' = '200', \n" +
                "  'fields.vc.min' = '10', \n" +
                "  'fields.vc.max' = '100'\n" +
                ")");

        //Sink
        tableEnv.executeSql("CREATE TABLE sinkTable (" +
                "    id       BIGINT,\n" +
                "    sid      STRING,\n" +
                "    ts   TIMESTAMP(3),\n" +
                "    flag   BOOLEAN,\n" +
                "    vc INT,\n" +
                "    dt STRING,\n" +
                "    dt1 STRING\n" +
                "   ) " +
                "   WITH (\n" +
                "  'connector' = 'print'" +
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable select *, DATE_FORMAT(`ts`, 'yyyy-MM-dd'), cast(CURRENT_DATE - INTERVAL '1' DAY as STRING) from sourceTable ");

        //check raw format
//        tableEnv.executeSql("CREATE TABLE sinkTable(sid STRING) WITH (\n" +
//                "  'connector' = 'kafka',\n" +
//                "  'topic' = 'water_sensor_1',\n" +
//                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
//                "  'sink.partitioner' = 'default',\n" +                 //默认default分区器，可选fixed，round-robin
//                "  'sink.parallelism' = '3',\n" +
//                "  'format' = 'raw'\n" +                               //raw格式序列化
//                ")");
//        tableEnv.executeSql("insert into sinkTable select sid from sourceTable");


    }
}
